package mqtt

import (
	"context"
	"errors"
	"sync"

	"micro-rulego/internal/broker"

	mqtt "github.com/eclipse/paho.mqtt.golang"
)

type mqttBroker struct {
	sync.Once
	sync.RWMutex

	connected bool
	addrs     []string
	client    mqtt.Client
	opts      broker.Options
	mqttOpts  *mqtt.ClientOptions

	closeCh chan error
}

type subscriber struct {
	t     mqtt.Token
	topic string
	opts  broker.SubscribeOptions
}

type publication struct {
	t   string
	err error
	m   []byte
	ctx context.Context
}

func (p *publication) Topic() string {
	return p.t
}

func (p *publication) Message() []byte {
	return p.m
}

func (p *publication) Context() context.Context {
	return p.ctx
}

func (p *publication) Ack() error {
	return nil
}

func (p *publication) Error() error {
	return p.err
}

func (s *subscriber) Options() broker.SubscribeOptions {
	return s.opts
}

func (s *subscriber) Topic() string {
	return s.topic
}

func (s *subscriber) Unsubscribe() error {
	if s.t != nil {
		return s.t.Error()
	}
	return nil
}

func (m *mqttBroker) Address() string {
	if len(m.addrs) > 0 {
		return m.addrs[0]
	}
	return ""
}

func (m *mqttBroker) Connect() error {
	m.Lock()
	defer m.Unlock()

	if m.connected {
		return nil
	}

	if m.client == nil {
		m.mqttOpts = mqtt.NewClientOptions()
		for _, addr := range m.addrs {
			m.mqttOpts.AddBroker(addr)
		}
		m.mqttOpts.SetClientID(m.opts.Name)
		if m.opts.TLSConfig != nil {
			m.mqttOpts.SetTLSConfig(m.opts.TLSConfig)
		}
		m.client = mqtt.NewClient(m.mqttOpts)
	}

	if token := m.client.Connect(); token.Wait() && token.Error() != nil {
		return token.Error()
	}

	m.connected = true
	return nil
}

func (m *mqttBroker) Disconnect() error {
	m.Lock()
	defer m.Unlock()

	if !m.connected {
		return nil
	}

	m.client.Disconnect(250)
	m.connected = false
	return nil
}

func (m *mqttBroker) Init(opts ...broker.Option) error {
	for _, o := range opts {
		o(&m.opts)
	}

	m.addrs = m.opts.Addrs
	if len(m.addrs) == 0 {
		m.addrs = []string{"tcp://127.0.0.1:1883"}
	}

	return nil
}

func (m *mqttBroker) Options() broker.Options {
	return m.opts
}

func (m *mqttBroker) Publish(topic string, msg []byte, opts ...broker.PublishOption) error {
	m.RLock()
	defer m.RUnlock()

	if !m.connected {
		return errors.New("not connected")
	}

	token := m.client.Publish(topic, 1, false, msg)
	return token.Error()
}

func (m *mqttBroker) Subscribe(topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
	m.RLock()
	defer m.RUnlock()

	if !m.connected {
		return nil, errors.New("not connected")
	}

	var options broker.SubscribeOptions
	for _, o := range opts {
		o(&options)
	}

	callback := func(client mqtt.Client, msg mqtt.Message) {
		pub := &publication{
			t:   msg.Topic(),
			m:   msg.Payload(),
			ctx: options.Context,
		}

		if err := handler(pub); err != nil {
			pub.err = err
			if m.opts.ErrorHandler != nil {
				m.opts.ErrorHandler(pub)
			}
		}
	}

	token := m.client.Subscribe(topic, 1, callback)
	if token.Wait() && token.Error() != nil {
		return nil, token.Error()
	}

	return &subscriber{
		t:     token,
		topic: topic,
		opts:  options,
	}, nil
}

func (m *mqttBroker) String() string {
	return "mqtt"
}

func NewBroker(opts ...broker.Option) broker.Broker {
	options := broker.Options{
		Context: context.Background(),
	}

	m := &mqttBroker{
		opts:    options,
		closeCh: make(chan error),
	}

	m.Init(opts...)
	return m
}
